package com.yunhe.history.service.data.impl;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Lists;
import com.yunhe.common.constant.KafkaConsts;
import com.yunhe.common.constant.MeasurementConsts;
import com.yunhe.common.constant.SystemConsts;
import com.yunhe.history.domain.data.ChargeAndDischargeCurve;
import com.yunhe.common.exception.ArgumentErrorException;
import com.yunhe.common.model.PageParam;
import com.yunhe.common.model.data.Measurement;
import com.yunhe.common.model.data.MeasurementDpuDTO;
import com.yunhe.common.model.msg.MeasurementMsgDTO;
import com.yunhe.history.service.data.MeasurementService;
import com.yunhe.common.util.*;
import com.yunhe.history.util.CurveUtils;
import com.yunhe.history.util.InfluxDBUtil;
import com.yunhe.common.util.CollectionUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.PageImpl;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;

/**
 * 量测Service
 * @author liuronglei
 */
@Service
public class InfluxDBMeasurementServiceImpl implements MeasurementService {
    private static final Logger logger = LoggerFactory.getLogger(InfluxDBMeasurementServiceImpl.class);
    private ObjectMapper mapper = JsonUtil.getObjectMapper();
    @Autowired
    private InfluxDBUtil influxDBUtil;
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 量测对象处理和判断
     * @param measurement 量测对象
     */
    private void processBeforeCreate(Measurement measurement) {
    }

    /**
     * 量测对象处理和判断（批量）
     * @param measurementList 量测对象列表
     */
    private void processBeforeCreate(List<Measurement> measurementList) {
    }

    /**
     * 新增量测对象
     * @param measurement 量测对象
     * @param frequency 时间频率
     */
    @Transactional
    public Measurement createMeasurement(Measurement measurement,
                                         String frequency)
            throws ArgumentErrorException {
        processBeforeCreate(measurement);
        influxDBUtil.putData(measurement, frequency);
        // 发送新增消息
        sendMeasurementRefresh(Collections.singletonList(measurement), frequency, SystemConsts.ACTION.ADD);
        return measurement;
    }

    /**
     * 批量新增量测对象
     * @param measurementList 量测对象列表
     * @param frequency 时间频率
     */
    @Transactional
    public Iterable<Measurement> createMeasurement(List<Measurement> measurementList,
                                                   String frequency)
            throws ArgumentErrorException {
        processBeforeCreate(measurementList);
        influxDBUtil.putData(measurementList, frequency);
        // 发送新增消息
        sendMeasurementRefresh(measurementList, frequency, SystemConsts.ACTION.ADD);
        return measurementList;
    }

    /**
     * 批量新增量测对象（异步方法）
     * @param measurementList 量测对象列表
     * @param frequency 时间频率
     */
    @Async
    @Transactional
    public Iterable<Measurement> createMeasurementAsync(List<Measurement> measurementList,
                                                        String frequency)
            throws ArgumentErrorException {
        return createMeasurement(measurementList, frequency);
    }

    /**
     * 新增DPU上传量测数据
     */
    @Async
    @Override
    public void createMeasurementFromDpuAsync(List<List<MeasurementDpuDTO>> measurementList) throws ArgumentErrorException {
        try {
            if (measurementList != null && measurementList.size() > 0) {
                List<Measurement> measurements = new ArrayList<>();
                for (List<MeasurementDpuDTO> list : measurementList) {
                    if (list != null && list.size() > 0) {
                        for (MeasurementDpuDTO measurementDpuDTO : list) {
                            Measurement measurement = new Measurement();
                            measurement.setPointNumber(measurementDpuDTO.getPointId());
                            measurement.setRtime(measurementDpuDTO.getTimestamp());
                            measurement.setVal(measurementDpuDTO.getMeasValue());
                            measurements.add(measurement);
                        }
                    }
                }
                if (measurements.size() > 0) {
                    createMeasurement(measurements, MeasurementConsts.FREQUENCY.Original.value());
                }
            }
        } catch (Exception e) {
            logger.error("DPU历史数据上传出错", e);
        }
    }

    /**
     * 根据点号删除量测
     * @param pointNumber 点号
     * @param frequency 时间频率
     */
    @Transactional
    public Long deleteByPointNumber(Long pointNumber,
                                    String frequency) {
        boolean suc = influxDBUtil.deleteAllRow(pointNumber, frequency);
        if (suc) {
            return 0L;
        } else {
            return -1L;
        }
    }

    /**
     * 获得点号和量测列表对应的Map
     * @param pointNumbers 点号
     * @param dtime 时间
     * @param frequency 时间频率
     * @param flag 标记
     * @param fill 是否填充开始和结束时间
     */
    @Override
    public Map<Long, List<Measurement>> getMeasurementMap(List<Long> pointNumbers,
                                                          List<String> dtime,
                                                          String frequency,
                                                          Integer flag,
                                                          Boolean fill,
                                                          Boolean queryAllAfterDtime) throws ArgumentErrorException {
        Map<Long, List<Measurement>> result = new HashMap<>(pointNumbers.size());
        if (dtime != null && dtime.size() == 1 && queryAllAfterDtime) {
            for (Long pointNumber : pointNumbers) {
                List<Measurement> measurements = influxDBUtil.getRowDataAfterDtime(pointNumber, dtime.get(0), frequency, flag);
                if (measurements != null) {
                    result.put(pointNumber, measurements);
                }
            }
        } else {
            for (Long pointNumber : pointNumbers) {
                Iterable<Measurement> list = findMeasurements(pointNumber, dtime, frequency, flag, fill, null);
                if (list != null) {
                    List<Measurement> measurements = Lists.newLinkedList(list);
                    result.put(pointNumber, measurements);
                }
            }
        }
        return result;
    }

    /**
     * 获得点号和量测列表对应的Map（分片方式）
     * @param pointNumbers 点号
     * @param dtime 时间
     * @param frequency 时间频率
     * @param flag 标记
     */
    @Override
    public Map<Long, List<Measurement>> getMeasurementMapByShard(List<Long> pointNumbers,
                                                                 List<String> dtime,
                                                                 String frequency,
                                                                 Integer flag,
                                                                 Boolean fill)
            throws ArgumentErrorException {
        if (CollectionUtil.isEmpty(pointNumbers) || CollectionUtil.isEmpty(dtime)) {
            throw new ArgumentErrorException("查询点号和时间不能为空");
        }
        Map<Long, List<Measurement>> result = new HashMap<>(pointNumbers.size());
        List<Measurement> scanners;
        if (dtime.size() == 2) {
            scanners = influxDBUtil.getResultsScanner(pointNumbers, dtime.get(0), dtime.get(1), frequency, flag);
        } else {
            scanners = influxDBUtil.getRowData(pointNumbers, dtime.get(0), frequency, flag);
        }
        // 组建点号和量测列表的映射
        for (Measurement measurement : scanners) {
            List<Measurement> oneList = result.computeIfAbsent(measurement.getPointNumber(), k -> new ArrayList<>());
            oneList.add(measurement);
        }
        // 开始填充开始和结束时间，为空时，默认为true
        // 只有原始值需要填充，只有时间范围需要填充
        if (MeasurementConsts.FREQUENCY.Original.equals(frequency) && (fill == null || fill) && dtime.size() == 2) {
            // 拿到点号列表在某个时间点的量测
            List<Measurement> aspectList = influxDBUtil.getRowData(pointNumbers, dtime.get(0), frequency, flag);
            // 组建成Map
            Map<Long, Measurement> aspectMap = new HashMap<>();
            if (CollectionUtil.isNotEmpty(aspectList)) {
                for (Measurement measurement : aspectList) {
                    aspectMap.put(measurement.getPointNumber(), measurement);
                }
            }
            // 循环每个点号，为点号填充开始和结束时间的数据
            for (Long pointNumber : pointNumbers) {
                // 从切面映射中取到切面量测
                Measurement firstMeas = aspectMap.get(pointNumber);
                if (firstMeas != null) {
                    List<Measurement> oneList = result.computeIfAbsent(pointNumber, v -> new ArrayList<>());
                    fillMeasurementEnds(oneList, pointNumber, frequency, flag, dtime.get(0), dtime.get(1), firstMeas);
                }
            }
        }
        return result;
    }

    /**
     * 根据条件获得量测列表
     * @param pointNumber 点号
     * @param dtime       时间
     * @param frequency 时间频率
     * @param flag 标记
     * @param fill 是否填充开始和结束时间
     * @param pageParam   分页对象
     */
    @Override
    public Iterable<Measurement> findMeasurements(Long pointNumber,
                                                  List<String> dtime,
                                                  String frequency,
                                                  Integer flag,
                                                  Boolean fill,
                                                  PageParam pageParam)
            throws ArgumentErrorException {
        if (pointNumber == null || dtime == null) {
            throw new ArgumentErrorException("查询点号和时间不能为空");
        }
        List<Measurement> resultList = new ArrayList<>();
        if (dtime.size() == 2) {
            resultList = influxDBUtil.getResultScanner(pointNumber, dtime.get(0), dtime.get(1), frequency, flag);
            // 开始填充开始和结束时间，为空时，默认为true
            // 只有原始值需要填充
            if (MeasurementConsts.FREQUENCY.Original.equals(frequency) && (fill == null || fill)) {
                resultList = fillMeasurementEnds(resultList, pointNumber, frequency, flag, dtime.get(0), dtime.get(1), null);
            }
        } else {
            Measurement measurement = influxDBUtil.getRowData(pointNumber, dtime.get(0), frequency, flag);
            if (measurement != null) {
                resultList.add(measurement);
            }
        }
        return resultList;
    }

    /**
     * 填充开始时间和结束时间两端的量测点
     * 正常情况下，只有原始值查询需要填充时调用
     * 关于firstMeas：如果已经可以拿到第一个量测点，则firstMeas直接传进来，否则传空即可
     * 关于firstMeas：这里有个性能问题，如果第一个量测本身就是null，则还是会查一次，这种情况一般不会出现，有机会可以优化
     * @param resultList 要填充的结果列表
     * @param pointNumber 点号
     * @param frequency 时间频率
     * @param flag 标记
     * @param startDtime 开始时间
     * @param endDtime 结束时间
     * @param firstMeas 第一个量测点
     */
    private List<Measurement> fillMeasurementEnds(List<Measurement> resultList,
                                                  Long pointNumber,
                                                  String frequency,
                                                  Integer flag,
                                                  String startDtime,
                                                  String endDtime,
                                                  Measurement firstMeas) {
        if (resultList == null) {
            resultList = new ArrayList<>();
        }
        // 如果结果数据是空，则开始时间和结束时间都填充成≤开始时间的第一个值
        if (resultList.size() == 0) {
            Measurement startDtimeMeas = firstMeas == null ? influxDBUtil.getRowData(pointNumber, startDtime, frequency, flag)
                    : firstMeas;
            if (startDtimeMeas != null) {
                Long currentRtime = System.currentTimeMillis();
                String timeZone = influxDBUtil.getTimeZoneByPointNumber(pointNumber);
                // 量测时间依旧要置为开始时间，但不能超过当前时间
                Long startRtime = DateTimeUtils.getTimestamp(startDtime, timeZone);
                if (startRtime <= currentRtime) {
                    startDtimeMeas.setRtime(startRtime);
                    startDtimeMeas.setDtime(startDtime);
                    resultList.add(startDtimeMeas);
                    Measurement endDtimeMeas = new Measurement();
                    endDtimeMeas.setPointNumber(pointNumber);
                    // 量测时间依旧要置为结束时间，但不能超过当前时间
                    Long endRtime = DateTimeUtils.getTimestamp(endDtime, timeZone);
                    if (endRtime > currentRtime) {
                        endRtime = currentRtime;
                        endDtime = DateTimeUtils.getDateString(endRtime, timeZone);
                    }
                    endDtimeMeas.setRtime(endRtime);
                    endDtimeMeas.setDtime(endDtime);
                    endDtimeMeas.setFlag(startDtimeMeas.getFlag());
                    endDtimeMeas.setVal(startDtimeMeas.getVal());
                    resultList.add(endDtimeMeas);
                }
            }
        // 如果存在结果数据，则分别填充≤开始时间和≤结束时间的第一个值
        } else {
            String timeZone = influxDBUtil.getTimeZoneByPointNumber(pointNumber);
            // 如果第一个值的时间和开始时间一致，则不需要填充开始时间
            // 如果末尾带毫秒肯定匹配不上，然后再用dtime进行判断
            if (resultList.get(0).getRtime() % 1000 != 0 || !startDtime.equals(resultList.get(0).getDtime())) {
                Measurement startDtimeMeas = firstMeas == null ? influxDBUtil.getRowData(pointNumber, startDtime, frequency, flag)
                        : firstMeas;
                if (startDtimeMeas != null) {
                    // 量测时间依旧要置为开始时间
                    startDtimeMeas.setRtime(DateTimeUtils.getTimestamp(startDtime, timeZone));
                    startDtimeMeas.setDtime(startDtime);
                    resultList.add(0, startDtimeMeas);
                }
            }
            Measurement lastMeas = resultList.get(resultList.size() - 1);
            Measurement endDtimeMeas = new Measurement();
            Long currentRtime = System.currentTimeMillis();
            // 结束时间不能超过当前时间
            Long endRtime = DateTimeUtils.getTimestamp(endDtime, timeZone);
            if (endRtime > currentRtime) {
                endRtime = currentRtime;
                endDtime = DateTimeUtils.getDateString(endRtime, timeZone);
            }
            // 只有结束时间＞查询结果最后一个量测时间，才填充结束时间
            if (endRtime > lastMeas.getRtime()) {
                endDtimeMeas.setPointNumber(pointNumber);
                endDtimeMeas.setRtime(endRtime);
                endDtimeMeas.setDtime(endDtime);
                endDtimeMeas.setFlag(lastMeas.getFlag());
                endDtimeMeas.setVal(lastMeas.getVal());
                resultList.add(endDtimeMeas);
            }
        }
        return resultList;
    }

    /**
     * 根据条件获得量测列表
     * @param pointNumbers 点号列表
     * @param dtime       时间
     * @param frequency 时间频率
     * @param flag 标记
     * @param pageParam   分页对象
     */
    @Override
    public Iterable<Measurement> findMeasurements(List<Long> pointNumbers,
                                                  List<String> dtime,
                                                  String frequency,
                                                  Integer flag,
                                                  PageParam pageParam) {
        if (QueryUtil.needPaging(pageParam)) {
            int page = pageParam.getPage();
            int size = pageParam.getSize();
            List<Measurement> content = new ArrayList<>();
            long count = influxDBUtil.getCount(pointNumbers, dtime, frequency, flag);
            if (dtime.size() == 1) {
                content = influxDBUtil.getRowData(pointNumbers, dtime.get(0), frequency, flag, page, size);
            } else if (dtime.size() > 1) {
                content = influxDBUtil.getRowData(pointNumbers, dtime, frequency, flag, page, size);
            }
            return new PageImpl<>(content != null ? content : new ArrayList<>(), QueryUtil.getPageRequest(pageParam), count);
        } else {
            if (dtime.size() == 1) {
                return influxDBUtil.getRowData(pointNumbers, dtime.get(0), frequency, flag);
            } else if (dtime.size() > 1) {
                return influxDBUtil.getRowData(pointNumbers, dtime, frequency, flag);
            }
        }
        return null;
    }

    @Override
    public void deleteMeasurements(Long pointNumber,
                                   List<String> dtime,
                                   String frequency,
                                   Integer flag,
                                   PageParam pageParam) throws ArgumentErrorException {
        if (pointNumber == null || dtime == null) {
            throw new ArgumentErrorException("查询点号和时间不能为空");
        }
        if (dtime.size() == 1) {
            influxDBUtil.deleteRow(pointNumber, dtime.get(0), frequency, flag);
        } else if (dtime.size() == 2) {
            influxDBUtil.deleteRow(pointNumber, dtime.get(0), dtime.get(1), frequency, flag);
        }
    }



    /**
     * 获得从开始时间后的所有量测列表
     * @param pointNumber 点号
     * @param stateDate 开始时间
     * @param frequency 时间频率
     * @param flag 标记
     */
    public Iterable<Measurement> findMeasurementsByStartDate(Long pointNumber,
                                                             String stateDate,
                                                             String frequency,
                                                             Integer flag) {
        List<Measurement> resultList = influxDBUtil.getResultByStartDate(pointNumber, stateDate, frequency, flag);
        // 开始填充开始和结束时间
        // 只有原始值需要填充
        if (MeasurementConsts.FREQUENCY.Original.equals(frequency)) {
            String timeZone = influxDBUtil.getTimeZoneByPointNumber(pointNumber);
            String endDate = DateTimeUtils.getDateString(System.currentTimeMillis(), timeZone);
            resultList = fillMeasurementEnds(resultList, pointNumber, frequency, flag, stateDate, endDate, null);
        }
        return resultList;
    }

//    /**
//     * 根据条件获得量测总数
//     * @param pointNumber 点号
//     * @param dtime 时间
//     */
//    public Long getCount(Long pointNumber,
//                         List<String> dtime)
//            throws ObjectNotFoundException {
//        if (pointNumber == null || dtime == null) {
//            throw new ObjectNotFoundException("查询点号和时间不能为空");
//        }
//        return influxDBUtil.getCount(pointNumber, dtime);
//    }

    public Iterable<Measurement> findMeasurementsEitherEnd(Long pointNumber,
                                                           List<String> dtime,
                                                           String frequency,
                                                           Integer flag) {
        if (dtime.size() < 2) {
            return null;
        } else {
            List<Measurement> result = new ArrayList<>();
            Measurement start = influxDBUtil.getResultScannerFirst(pointNumber, dtime.get(0), dtime.get(1), frequency, flag);
            Measurement end = influxDBUtil.getResultScannerLast(pointNumber, dtime.get(0), dtime.get(1), frequency, flag);
            if (start != null && end != null) {
                result.add(start);
                result.add(end);
            }
            return result;
        }
    }

    @Override
    public List<List<Measurement>> findMeasurementsEitherEnd(List<Long> pointNumber, List<String> dtime, String frequency, Integer flag) {
        List<List<Measurement>> result = new ArrayList<>();
        for (Long pid : pointNumber) {
            List<Measurement> list = (List<Measurement>) findMeasurementsEitherEnd(pid, dtime, frequency, flag);
            result.add(list);
        }
        return result;
    }

    /**
     * 获得点号最新的值
     * @param pointNumber 点号
     */
    @Override
    public Measurement findMeasurementsLast(Long pointNumber) {
        return influxDBUtil.getResultScannerLast(pointNumber);
    }

    /**
     * 批量获得点号最新的值
     * @param pointNumbers 点号
     * @param frequency 时间频率
     * @param flag 标记
     */
    @Override
    public Map<Long, Measurement> findMeasurementsLastBatch(List<Long> pointNumbers,
                                                            String frequency,
                                                            Integer flag,
                                                            String endTime) {
        return influxDBUtil.getResultScannerLast(pointNumbers, frequency, flag, endTime);
    }

//    public List<String> findPnWithoutValue(String dtime) {
//        Map<Long, Object> hasValuePoint = influxDBUtil.getPointNumberByDtime(dtime);
//        List<String> result = new ArrayList<>();
//        for (Map.Entry<Long, Object> entry : hasValuePoint.entrySet()) {
//            result.add(String.valueOf(entry.getKey()));
//        }
//        return result;
//    }

//    /**
//     * 获得指定点号和时间下的量测值
//     * @param pointNumber 点号
//     * @param dtime    时间
//     */
//    @Override
//    public Measurement findCustomValue(Long pointNumber, String dtime) throws ObjectNotFoundException {
//        return influxDBUtil.getRowData(pointNumber, dtime);
//    }

//    /**
//     * 获得点号的最新量测值
//     * @param pointNumber 点号
//     */
//    @Override
//    public Measurement findLastValue(Long pointNumber) throws ObjectNotFoundException {
//        Measurement currentVal = findMeasurementsLast(pointNumber);
//        if (currentVal == null) {
//            currentVal = influxDBUtil.getPreValue(pointNumber, null);
//        }
//        return currentVal;
//    }

    /**
     * 根据点号获得点号与最后一次值不同的前一次的值
     * @param pointNumber 点号
     */
    public Double getPreValue(Long pointNumber) {
        Measurement currentVal = findMeasurementsLast(pointNumber);
        if (currentVal == null) {
            currentVal = influxDBUtil.getPreValue(pointNumber, null);
        }
        currentVal = influxDBUtil.getPreValue(pointNumber, currentVal.getVal());
        if (currentVal != null) {
            return currentVal.getVal();
        } else {
            return null;
        }
    }

    @Override
    public Double getPreValue(Long pointNumber,
                              String dtime,
                              String frequency,
                              Integer flag) {
        Measurement preValue = influxDBUtil.getPreValueByTime(pointNumber, dtime, frequency, flag);
        if (preValue != null) {
            return preValue.getVal();
        } else {
            return null;
        }
    }

    @Override
    public Map<Long, Double> getPreValues(List<Long> pointNumbers,
                                          Long rtime,
                                          String frequency,
                                          Integer flag) {
        Map<Long, Double> map = new HashMap<>();
        if (CollectionUtil.isNotEmpty(pointNumbers)) {
            for (Long pointNumber : pointNumbers) {
                Measurement preValue = influxDBUtil.getPreValueByRtime(pointNumber, rtime, frequency, flag);
                if (preValue != null) {
                    map.put(pointNumber, preValue.getVal());
                } else {
                    map.put(pointNumber, null);
                }
            }
        }
        return map;
    }

    @Override
    public Measurement getNextMeasurement(Long pointNumber,
                                          String dtime,
                                          String frequency,
                                          Integer flag) {
        Measurement nextValue = influxDBUtil.getNextValueByTime(pointNumber, dtime, frequency, flag);
        if (nextValue != null) {
            return nextValue;
        } else {
            return null;
        }
    }

    @Override
    public ChargeAndDischargeCurve calcByIntegral(Long pointNumber, List<String> dates) throws ArgumentErrorException {
        List<Measurement> measurements = Lists.newArrayList(this.findMeasurements(pointNumber, dates, MeasurementConsts.FREQUENCY.Original.value(), null, false,
                new PageParam()));
        return CurveUtils.getChargeAndDischargeCurve(measurements, dates.get(0));
    }

    /**
     * 根据条件获得一段时间内量测值最大的量测点
     * @param pointNumbers 点号列表
     * @param dtime 时间范围
     * @param frequency 时间频率
     * @param flag 标记
     */
    @Override
    public List<Measurement> getMaxMeasurement(List<Long> pointNumbers, List<String> dtime, String frequency, Integer flag) throws ArgumentErrorException {
        if (CollectionUtil.isEmpty(pointNumbers) || CollectionUtil.isEmpty(dtime) || dtime.size() < 2) {
            throw new ArgumentErrorException("查询点号和时间段不能为空");
        }
        return influxDBUtil.getMaxValueByTime(pointNumbers, dtime.get(0), dtime.get(1), frequency, flag);
    }

    /**
     * 根据开始时间和结束时间获得量测值（没有对应时间点的数据，取上一个点）
     */
    @Override
    public List<List<Measurement>> findByStartAndEndTime(List<Long> pointNumbers, List<String> dtime, String frequency, Integer flag) {
        Map<Long, Measurement> startValues = influxDBUtil.getResultScannerLast(pointNumbers, frequency, flag, dtime.get(0));
        Map<Long, Measurement> endValues = influxDBUtil.getResultScannerLast(pointNumbers, frequency, flag, dtime.get(1));
        List<List<Measurement>> result = new ArrayList<>();
        for (Long pid : pointNumbers) {
            List<Measurement> list = new ArrayList<>();
            if (startValues != null && startValues.get(pid) != null) {
                list.add(startValues.get(pid));
            }
            if (endValues != null && endValues.get(pid) != null) {
                list.add(endValues.get(pid));
            }
            result.add(list);
        }
        return result;
    }

    /**
     * 发送历史量测事件更新消息
     * @param measurements 量测列表
     * @param frequency 时间频率
     * @param action 事件类型
     */
    private void sendMeasurementRefresh(List<Measurement> measurements, String frequency, SystemConsts.ACTION action) {
        if (CollectionUtil.isNotEmpty(measurements)) {
            MeasurementMsgDTO msg = new MeasurementMsgDTO();
            msg.setAction(action.value());
            msg.setFrequency(frequency);
            msg.setMeasurements(measurements);
            try {
                kafkaTemplate.send(KafkaConsts.Topic.MeasurementRefresh, mapper.writeValueAsString(msg));
            } catch (Exception e) {
                logger.error("Kafka发送历史量测更新消息失败：{}" + measurements.toString());
                e.printStackTrace();
            }
        }
    }

}
